package com.fulihui.wgj.biz.impl;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.fulihui.common.util.JedisUtil;
import com.fulihui.weixinsdk.model.TemplateItem;
import com.fulihui.wgj.biz.SendTemplateMsgBiz;
import com.fulihui.wgj.biz.WxUserInfoBiz;
import com.fulihui.wgj.dal.impala.domain.WxUserInfoSend;
import com.fulihui.wgj.dal.mybatis.domain.WxSendTemplateLog;
import com.fulihui.wgj.dal.mybatis.repository.WxSendTemplateLogRepository;
import com.fulihui.wgj.facade.dto.MsgUserDTO;
import com.fulihui.wgj.facade.dto.TemplateMsgDTO;
import com.fulihui.wgj.facade.msg.TemplateMsg;
import com.fulihui.wgj.facade.request.template.CalcSendCountRequest;
import com.fulihui.wgj.mq.kafka.producer.KafkaProducer;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * 模板发送业务
 *
 * @author wang_wx
 * @date 2018-07-27
 */
@Service
public class SendTemplateMsgBizImpl implements SendTemplateMsgBiz {

    private static final Logger LOGGER = LoggerFactory.getLogger(SendTemplateMsgBizImpl.class);

    /**
     * 队列批次长度--一般
     */
    private static final int MQ_LENGTH_GENERALLY = 200;

    /**
     * 模板发送到kafka队列消息redis_key
     */
    private static final String WX_TEMPLATE_SEND_REDIS_PREFIX = "TEMPLATE_SEND_";

    @Autowired
    ThreadPoolTaskExecutor threadPool;

    @Autowired
    JedisUtil jedisUtil;

    @Autowired
    KafkaProducer kafkaProducer;

    @Autowired
    WxUserInfoBiz wxUserInfoBiz;

    @Autowired
    WxSendTemplateLogRepository wxSendTemplateLogRepository;

    @Override
    public void sendTemplateMsg(CalcSendCountRequest request, WxSendTemplateLog task) {
        LOGGER.info("sendTemplateMsg start .......");
        Integer recordCount = task.getAllTotal();
        if (recordCount == null || recordCount == 0) {
            recordCount = Math.toIntExact(wxUserInfoBiz.queryCount(request));
        }
        // 发送的模板数据
        StringBuilder sendRedisKey = new StringBuilder(WX_TEMPLATE_SEND_REDIS_PREFIX);
        sendRedisKey.append(request.getAppId()).append("_").append(task.getId());

        TemplateMsgDTO templateMsgDTO = new TemplateMsgDTO();
        templateMsgDTO.setTemplateId(task.getTemplateId());
        templateMsgDTO.setTemplateUrl(task.getJumpUrl());
        templateMsgDTO.setMiniProgramAppId(task.getMiniAppId());
        templateMsgDTO.setMiniProgramUrl(task.getMiniProgramUrl());
        // 模板发送数据
        String templateContent = request.getTemplateContent();
        Map<String, TemplateItem> data = JSON.parseObject(templateContent,
                new TypeReference<Map<String, TemplateItem>>() {
                });
        templateMsgDTO.setData(data);
        jedisUtil.set(sendRedisKey.toString(), JSON.toJSONString(templateMsgDTO), 604800);

        //异步执行
        Integer finalRecordCount = recordCount;
        LOGGER.info("#####发送模板消息的发送总数是" + finalRecordCount);
        threadPool.execute(() -> {
            try {
                //要发送的json
                TemplateMsg templateMsg = new TemplateMsg();
                templateMsg.setIsLast(false);
                templateMsg.setAppId(request.getAppId());
                templateMsg.setRecordId(task.getId());
                templateMsg.setNewsKey(sendRedisKey.toString());
                LOGGER.info("发送模板消息,:Condition{}", request);

                //分页查询
                int pageSize = 5000;
                long pageCount = (finalRecordCount + pageSize - 1) / pageSize;
                LOGGER.info(String.format("#####发送模板消息的发送总数是：%d 发送模板的分页是：%d 发送模板的批次是 %d发送模板的公众号是：%s", finalRecordCount, pageCount, task.getId(), request.getAppId()));
                String nextOpenid = null;
                boolean isLastPage = false;
                //发送总条数
                for (int pageNum = 1; pageNum <= pageCount; pageNum++) {
                    if (pageNum == pageCount) {
                        isLastPage = true;
                    }
                    try {
                        //分批取出用户信息
                        request.setPage(pageNum);
                        request.setRows(pageSize);
                        List<WxUserInfoSend> userInfoSends = wxUserInfoBiz.queryPage(request, nextOpenid);

                        //缓解查询压力，每批休眠
                        TimeUnit.SECONDS.sleep(3);

                        if (!CollectionUtils.isEmpty(userInfoSends)) {
                            try {
                                putTemplateMsg2MQ(userInfoSends, templateMsg, isLastPage);
                            } catch (Exception e) {
                                LOGGER.error("往kafka消息队列放数据出错:{}", e);
                            }
                        }

                        nextOpenid = userInfoSends.get(userInfoSends.size() - 1).getOpenid();
                    } catch (Exception e) {
                        LOGGER.error("发送模板消息内部错误,:e{}", e);
                    }
                }
            } catch (Exception e) {
                LOGGER.error("发送模板消息错误,:e{}", e);
            }
        });
    }

    @Override
    public void sendTemplateMsg4Game(CalcSendCountRequest request, WxSendTemplateLog task) {
        LOGGER.info("sendTemplateMsg4Game start .......");
        Integer recordCount = task.getAllTotal();
        if (recordCount == null || recordCount == 0) {
            recordCount = Math.toIntExact(wxUserInfoBiz.queryCount4Game(request));
        }
        // 发送的模板数据
        StringBuilder sendRedisKey = new StringBuilder(WX_TEMPLATE_SEND_REDIS_PREFIX);
        sendRedisKey.append(request.getAppId()).append("_").append(task.getId());

        TemplateMsgDTO templateMsgDTO = new TemplateMsgDTO();
        templateMsgDTO.setTemplateId(task.getTemplateId());
        templateMsgDTO.setTemplateUrl(task.getJumpUrl());
        templateMsgDTO.setMiniProgramAppId(task.getMiniAppId());
        templateMsgDTO.setMiniProgramUrl(task.getMiniProgramUrl());
        // 模板发送数据
        String templateContent = request.getTemplateContent();
        Map<String, TemplateItem> data = JSON.parseObject(templateContent,
                new TypeReference<Map<String, TemplateItem>>() {
                });
        templateMsgDTO.setData(data);
        jedisUtil.set(sendRedisKey.toString(), JSON.toJSONString(templateMsgDTO), 604800);

        //异步执行
        Integer finalRecordCount = recordCount;
        LOGGER.info("#####游戏发送模板消息的发送总数是" + finalRecordCount);
        threadPool.execute(() -> {
            try {
                //要发送的json
                TemplateMsg templateMsg = new TemplateMsg();
                templateMsg.setIsLast(false);
                templateMsg.setAppId(request.getAppId());
                templateMsg.setRecordId(task.getId());
                templateMsg.setNewsKey(sendRedisKey.toString());
                LOGGER.info("游戏发送模板消息,:Condition{}", request);

                //分页查询
                int pageSize = 5000;
                long pageCount = (finalRecordCount + pageSize - 1) / pageSize;
                LOGGER.info(String.format("#####游戏发送模板消息的发送总数是：%d 发送模板的分页是：%d 发送模板的批次是 %d发送模板的公众号是：%s", finalRecordCount, pageCount, task.getId(), request.getAppId()));
                String nextOpenid = null;
                boolean isLastPage = false;
                //发送总条数
                for (int pageNum = 1; pageNum <= pageCount; pageNum++) {
                    if (pageNum == pageCount) {
                        isLastPage = true;
                    }
                    try {
                        //分批取出用户信息
                        request.setPage(pageNum);
                        request.setRows(pageSize);
                        List<WxUserInfoSend> wxUserInfoSends = wxUserInfoBiz.queryPage4Game(request, nextOpenid);

                        //缓解查询压力，每批休眠
                        TimeUnit.SECONDS.sleep(3);

                        if (!CollectionUtils.isEmpty(wxUserInfoSends)) {
                            try {
                                putTemplateMsg2MQ(wxUserInfoSends, templateMsg, isLastPage);
                            } catch (Exception e) {
                                LOGGER.error("往kafka消息队列放数据出错:{}", e);
                            }
                        }

                        nextOpenid = wxUserInfoSends.get(wxUserInfoSends.size() - 1).getOpenid();
                    } catch (Exception e) {
                        LOGGER.error("发送模板消息内部错误,:e{}", e);
                    }
                }
            } catch (Exception e) {
                LOGGER.error("发送模板消息错误,:e{}", e);
            }
        });
    }

    /**
     * 将模板消息推送到kafka队列
     *
     * @param userInfoSends
     * @param templateMsg
     * @param isLastPage
     */
    private void putTemplateMsg2MQ(List<WxUserInfoSend> userInfoSends, TemplateMsg templateMsg, boolean isLastPage) {
        TemplateMsg templateMsg_ = new TemplateMsg();
        BeanUtils.copyProperties(templateMsg, templateMsg_);
        templateMsg_.setIsLast(isLastPage);

        // 不足200条的直接扔队列
        if (userInfoSends.size() <= MQ_LENGTH_GENERALLY) {
            List<MsgUserDTO> msgUserDTOS = Lists.newArrayList();
            userInfoSends.stream().forEach(wxUserInfoSend -> {
                MsgUserDTO msgUserDTO = new MsgUserDTO();
                msgUserDTO.setOpenId(wxUserInfoSend.getOpenid());
                msgUserDTO.setNickName(wxUserInfoSend.getNickname());
                msgUserDTOS.add(msgUserDTO);
            });
            templateMsg_.setUsers(msgUserDTOS);
            kafkaProducer.sendTemplateMsg(templateMsg_);
        } else {
            // 按200条分割一份 扔队列
            List<List<WxUserInfoSend>> partition = Lists.partition(userInfoSends, MQ_LENGTH_GENERALLY);
            partition.stream().forEach(infoDTOS -> {
                List<MsgUserDTO> msgUserDTOS = Lists.newArrayList();
                infoDTOS.stream().forEach(wxUserInfoSend -> {
                    MsgUserDTO msgUserDTO = new MsgUserDTO();
                    msgUserDTO.setOpenId(wxUserInfoSend.getOpenid());
                    msgUserDTO.setNickName(wxUserInfoSend.getNickname());
                    msgUserDTOS.add(msgUserDTO);
                });
                templateMsg_.setUsers(msgUserDTOS);
                kafkaProducer.sendTemplateMsg(templateMsg_);
            });
        }
    }

    @Override
    public void sendTemplateMsg4Local(List<String> openIds, WxSendTemplateLog templateLog) {
        LOGGER.info("sendTemplateMsg4Local start .......");
        // 推送人数为0 不进行推送
        if (CollectionUtils.isEmpty(openIds)) {
            return;
        }
        int recordCount = openIds.size();
        LOGGER.info("sendTemplateMsg4Local recordCount:{}", recordCount);

        // 发送的模板数据
        StringBuilder sendRedisKey = new StringBuilder(WX_TEMPLATE_SEND_REDIS_PREFIX);
        sendRedisKey.append(templateLog.getAppId()).append("_").append(templateLog.getId());

        TemplateMsgDTO templateMsgDTO = new TemplateMsgDTO();
        templateMsgDTO.setTemplateId(templateLog.getTemplateId());
        templateMsgDTO.setTemplateUrl(templateLog.getJumpUrl());
        templateMsgDTO.setMiniProgramAppId(templateLog.getMiniAppId());
        templateMsgDTO.setMiniProgramUrl(templateLog.getMiniProgramUrl());
        // 模板发送数据
        String templateContent = templateLog.getTemplateData();
        Map<String, TemplateItem> data = JSON.parseObject(templateContent,
                new TypeReference<Map<String, TemplateItem>>() {
                });
        templateMsgDTO.setData(data);
        jedisUtil.set(sendRedisKey.toString(), JSON.toJSONString(templateMsgDTO), 604800);

        //异步执行
        Integer finalRecordCount = recordCount;
        LOGGER.info("#####本地模板消息的发送总数是" + finalRecordCount);
        threadPool.execute(() -> {
            try {
                //要发送的json
                TemplateMsg templateMsg = new TemplateMsg();
                templateMsg.setAppId(templateLog.getAppId());
                templateMsg.setRecordId(templateLog.getId());
                templateMsg.setNewsKey(sendRedisKey.toString());

                // 推送到队列
                putTemplateMsg2MQ4Local(openIds, templateMsg);

            } catch (Exception e) {
                LOGGER.error("发送本地模板消息错误,:e{}", e);
            }
        });
    }

    /**
     * 将模板消息推送到kafka队列 本地
     *
     * @param openIds
     * @param templateMsg
     */
    private void putTemplateMsg2MQ4Local(List<String> openIds, TemplateMsg templateMsg) {
        TemplateMsg templateMsg_ = new TemplateMsg();
        BeanUtils.copyProperties(templateMsg, templateMsg_);

        // 按200条分割一份 扔队列
        List<List<String>> partition = Lists.partition(openIds, MQ_LENGTH_GENERALLY);
        for (int i = 0; i < partition.size(); i++) {
            templateMsg_.setIsLast(Boolean.FALSE);
            if (i == partition.size() - 1) {
                // 最后一份
                templateMsg_.setIsLast(Boolean.TRUE);
            }
            List<String> openIdList = partition.get(i);
            List<MsgUserDTO> msgUserDTOS = Lists.newArrayList();
            openIdList.stream().forEach(openid -> {
                MsgUserDTO msgUserDTO = new MsgUserDTO();
                msgUserDTO.setOpenId(openid);
                msgUserDTOS.add(msgUserDTO);
            });
            templateMsg_.setUsers(msgUserDTOS);
            kafkaProducer.sendTemplateMsg(templateMsg_);
        }
    }
}
